package kshell.cn.msg.service.impl;

import kshell.cn.msg.service.ProducerService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;

import java.util.concurrent.locks.LockSupport;
import java.util.function.Supplier;

/**
 * 生产者实现
 */
@Service
@Slf4j
public class ProducerServiceImpl implements ProducerService {
    private final Sinks.Many<Message<String>> sinks =
            Sinks.many().multicast().onBackpressureBuffer();

    @Bean
    public Supplier<Flux<Message<String>>> source() {
        return sinks::asFlux;
    }

        /**
         * 发送消息
         *
         * @param message
         */
    @Override
    public void send(String message) {
            log.info("生产者发送消息:{}", message);
            Message<String> msg = MessageBuilder.withPayload(message).build();
            while (sinks.tryEmitNext(msg).isFailure()) {
                LockSupport.parkNanos(10);
            }
    }
}
